package rpc_protobuf

import (
	"context"
	"fmt"
	"goland-study/go_code/concurrency_design_model/pubsub/impl1"
	"google.golang.org/grpc"
	"io"
	"net"
	"strings"
	"time"
)

// gRPC是Google公司基于Protobuf开发的跨语言的开源RPC框架。
// gRPC基于HTTP/2协议设计，可以基于一个HTTP/2链接提供多个服务，对于移动设备更加友好
// ------------ 4.4.1 grpc 技术栈 --------------
// 链接: https://books.studygolang.com/advanced-go-programming-book/ch4-rpc/ch4-04-grpc.html
//    application
//   Generated Stubs
//  gRpc go core + interceptors
//  http/2; Security(TLS/SSL or ALTS,ETC)
//  Unix Domain Sockets  |||   TCP
//
// 最底层为TCP或Unix Socket协议，在此之上是HTTP/2协议的实现，然后在HTTP/2协议之上又构建了针对Go语言的gRPC核心库。
// 应用程序通过gRPC插件生产的Stub代码和gRPC核心库通信，也可以直接和gRPC核心库通信

// -------------- 4.4.2 grpc入门----------
// gRPC通过context.Context参数，为每个方法调用提供了上下文支持。
// 客户端在调用方法的时候，可以通过可选的grpc.CallOption类型的参数提供额外的上下文信息

type HelloServiceImpl struct{}

func (p *HelloServiceImpl) Hello(ctx context.Context, args *StringGrpc) (*StringGrpc, error) {
	return &StringGrpc{Value: "hello:" + args.GetValue()}, nil
}

// 服务端在循环中接收客户端发来的数据，如果遇到io.EOF表示客户端流被关闭，如果函数退出表示服务端流关闭。
// 生成返回的数据通过流发送给客户端，双向流数据的发送和接收都是完全独立的行为
func (*HelloServiceImpl) Channel(stream HelloGrpcService_ChannelServer) error {
	for {
		args, err := stream.Recv()
		if err != nil {
			if err == io.EOF {
				return nil
			}
			return err
		}

		err = stream.SendMsg(&StringGrpc{Value: "stream hello:" + args.GetValue()})
		if err != nil {
			return err
		}
	}
}

// 启动grpcServer
func GrpcHelloServer() {
	grpcServer := grpc.NewServer()
	RegisterHelloGrpcServiceServer(grpcServer, new(HelloServiceImpl))

	listener, err := net.Listen("tcp", ":1234")
	if err != nil {
		panic(err)
	}

	err = grpcServer.Serve(listener)
	if err != nil {
		panic(err)
	}
}

type PubSubServiceServerImpl struct {
	pub *impl1.Publisher
}

func NewPubSubServiceServerImpl() *PubSubServiceServerImpl {
	return &PubSubServiceServerImpl{pub: impl1.NewPublisher(100*time.Millisecond, 10)}
}

func (s *PubSubServiceServerImpl) Publish(ctx context.Context, subString *PubSubString) (*PubSubString, error) {
	s.pub.Publish(subString)
	return &PubSubString{}, nil
}

func (s *PubSubServiceServerImpl) Subscribe(subString *PubSubString, server PubSubService_SubscribeServer) error {
	cn := s.pub.SubscribeTopic(func(v interface{}) bool {
		if key, ok := v.(string); ok {
			if strings.Contains(key, subString.GetValue()) {
				return true
			}
		}
		return false
	})

	for v := range cn {
		if err := server.Send(&PubSubString{Value: v.(string)}); err != nil {
			return err
		}
	}

	return nil
}

// grpcClient启动
// gRPC和标准库的RPC框架有一个区别，gRPC生成的接口并不支持异步调用。
// 不过我们可以在多个Goroutine之间安全地共享gRPC底层的HTTP/2链接，
// 因此可以通过在另一个Goroutine阻塞调用的方式模拟异步调用。
func GrpcHelloClient() {
	// grpc.Dial负责和gRPC服务建立链接
	//conn, err := grpc.Dial("localhost:1234", grpc.WithTransportCredentials(insecure.NewCredentials()))
	//if err != nil {
	//	panic(err)
	//}
	//defer conn.Close()
	//
	//// 基于已经建立的链接构造HelloServiceClient对象
	//serviceClient := NewHelloGrpcServiceClient(conn)
	//// 返回的client其实是一个HelloServiceClient接口对象，通过接口定义的方法就可以调用服务端对应的gRPC服务提供的方法
	//reply, err := serviceClient.Hello(context.Background(), &StringGrpc{Value: "hello"})
	//if err != nil {
	//	panic(err)
	//}
	//
	//fmt.Println(reply.GetValue())
}

// --------------- grpc流 -------------------
// RPC是远程函数调用，因此每次调用的函数参数和返回值不能太大，否则将严重影响每次调用的响应时间
// 因此传统的RPC方法调用对于上传和下载较大数据量场景并不适合
// 时传统RPC模式也不适用于对时间不确定的订阅和发布模式
// 为此，gRPC框架针对服务器端和客户端分别提供了流特性
func GrpcStreamHelloServer() {
	grpcServer := grpc.NewServer()
	RegisterHelloGrpcServiceServer(grpcServer, new(HelloServiceImpl))

	listener, err := net.Listen("tcp", ":1234")
	if err != nil {
		panic(err)
	}

	err = grpcServer.Serve(listener)
	if err != nil {
		panic(err)
	}
}

func GrpcStreamHelloClient() {
	// grpc.Dial负责和gRPC服务建立链接;grpc版本不兼容,去掉
	//conn, err := grpc.Dial("localhost:1234", grpc.WithTransportCredentials(insecure.NewCredentials()))
	//if err != nil {
	//	panic(err)
	//}
	//defer conn.Close()
	//
	//stream, err := NewHelloGrpcServiceClient(conn).Channel(context.Background())
	//if err != nil {
	//	panic(err)
	//}

	//go func() {
	//	for i := 0; i < 6; i++ {
	//		err = stream.SendMsg(&StringGrpc{Value: "hello" + strconv.Itoa(i)})
	//		if err != nil {
	//			if err == io.EOF {
	//				break
	//			}
	//			panic(err)
	//		}
	//		time.Sleep(time.Second * 1)
	//	}
	//
	//	stream.CloseSend()
	//}()
	//
	//go func() {
	//	for {
	//		//reply, err := stream.CloseAndRecv()
	//		//if err != nil {
	//		//	if err == io.EOF {
	//		//		break
	//		//	}
	//		//	panic(err)
	//		//}
	//		var reply StringGrpc
	//		err = stream.RecvMsg(&reply)
	//		if err != nil {
	//			if err == io.EOF {
	//				break
	//			}
	//			panic(err)
	//		}
	//		fmt.Println(reply.GetValue())
	//	}
	//}()

	//time.Sleep(time.Second * 60)

	//for i := 0; i < 5; i++ {
	//	req := &StringGrpc{
	//		Value: fmt.Sprintf("client message %d", i),
	//	}
	//
	//	if err := stream.Send(req); err != nil {
	//		log.Fatalf("Failed to send message to server: %v", err)
	//	}
	//
	//	var resp StringGrpc
	//	err = stream.RecvMsg(&resp)
	//	if err == io.EOF {
	//		break
	//	}
	//	if err != nil {
	//		log.Fatalf("Failed to receive message from server: %v", err)
	//	}
	//
	//	fmt.Printf("Received message from server: %s\n", resp.GetValue())
	//}
	//
	//if err := stream.CloseSend(); err != nil {
	//	log.Fatalf("Failed to close stream: %v", err)
	//}
}

// --- 发布订阅模式 ----
func DemoSubPub() {
	p := impl1.NewPublisher(100*time.Millisecond, 10)
	golang := p.SubscribeTopic(func(v interface{}) bool {
		if key, ok := v.(string); ok {
			if strings.HasPrefix(key, "golang") {
				return true
			}
		}
		return false
	})
	docker := p.SubscribeTopic(func(v interface{}) bool {
		if key, ok := v.(string); ok {
			if strings.HasPrefix(key, "docker") {
				return true
			}
		}
		return false
	})
	go p.Publish("golang:这本书真垃圾")
	go p.Publish("docker:还引入一个docker依赖")
	go p.Publish("hello")
	time.Sleep(1 * time.Second)

	go func() {
		fmt.Printf("golang的收得到数据: %v\n", <-golang)
	}()
	go func() {
		fmt.Printf("docker的收得到数据: %v\n", <-docker)
	}()

	//<-make(chan bool)
}

//-- 发布订阅模式优化
